package job.gfp.ranks_v2.init;

import java.io.IOException;
import java.util.List;

import job.gfp.ranks_v2.RanksV2_Algorithm;
import job.gfp.ranks_v2.RanksV2_Data;
import job.gfp.ranks_v2.RanksV2_Util;
import mapred.base.TextMapper;

import org.apache.hadoop.io.Text;

import util.FeatureMap;

public class RanksV2_Initialize_Mapper extends TextMapper {

    private static List<RanksV2_Algorithm> algorithms;

    @Override
    public void map(Text key, Text value) throws IOException {
        String source = key.toString();
        String v = value.toString();
        FeatureMap targets = FeatureMap.parse(v, RanksV2_Util.DELIM);
        StringBuilder sb = new StringBuilder();
        for (RanksV2_Algorithm algorithm : algorithms) {
            RanksV2_Data data = algorithm.initialize(source, targets, this
                    .getJobConfig());
            sb.append(" ");
            sb.append(data.toString());
        }

        StringBuilder sbValue = new StringBuilder(sb.substring(1));
        // TODO Is this necessary?
        sb.delete(0, sb.length());
        sbValue.append("\t");
        sbValue.append(v);
        this.output(key, sbValue.toString());
    }

    @Override
    public void setUp() {
        algorithms = RanksV2_Util.loadAlgorithms(this.getConfig("algorithms"));
    }

    @Override
    public void tearDown() throws IOException {
    }

}
